package org.ring.common.transport;

import io.netty.bootstrap.ServerBootstrap;
import io.netty.channel.Channel;
import io.netty.channel.ChannelDuplexHandler;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelFutureListener;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.ChannelOption;
import io.netty.channel.ChannelPipeline;
import io.netty.channel.ChannelPromise;
import io.netty.channel.EventLoopGroup;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.nio.NioServerSocketChannel;
import io.netty.handler.timeout.ReadTimeoutHandler;

import org.ring.common.URL;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.concurrent.CompletableFuture;

/**

 * @since 2023-10-26 18:41
 */
public class TcpServerTransport {
    private final static Logger logger = LoggerFactory.getLogger(TcpServerTransport.class);
    protected Channel serverChannel = null;

    private EventLoopGroup bossGroup;
    private EventLoopGroup workerGroup;
    private ServerBootstrap serverBootstrap = null;

    private ConnectionManager connectionManager = new ConnectionManager(false);

    private HandlerRegistry handlerRegistry;

    private URL url;

    public TcpServerTransport(URL url){
        this.url = url;
    }

    public TcpServerTransport handlerRegistry(HandlerRegistry handlerRegistry){
        this.handlerRegistry = handlerRegistry;
        return this;
    }

    public ConnectionManager getConnectionManager() {
        return connectionManager;
    }

    public void start(){

        this.bossGroup = new NioEventLoopGroup(1);
        this.workerGroup = new NioEventLoopGroup();


        this.serverBootstrap = new ServerBootstrap()
            .group(bossGroup, workerGroup)
                .channel(NioServerSocketChannel.class)
                .childHandler(initChildChannelHandler());

        serverBootstrap.option(ChannelOption.SO_BACKLOG, 4096);
        serverBootstrap.childOption(ChannelOption.SO_KEEPALIVE, true);
        serverBootstrap.childOption(ChannelOption.TCP_NODELAY, true);
        bind();
    }


    private void bind() {
        int port = url.getPort();
        ChannelFuture channelFuture = serverBootstrap.bind(port);
        channelFuture.syncUninterruptibly();
        this.serverChannel = channelFuture.channel();
        this.serverChannel.closeFuture().addListener(new ChannelFutureListener() {
            @Override
            public void operationComplete(ChannelFuture future) throws Exception {
                bossGroup.shutdownGracefully();
                workerGroup.shutdownGracefully();
            }
        });
        logger.info("Tcp Server listening on port " + port);
    }

    public void shutdown() {
        if (serverChannel != null) {
            serverChannel.close();
        }
    }

    protected ChannelInitializer<Channel> initChildChannelHandler() {
        return new ChannelInitializer<Channel>() {
            @Override
            protected void initChannel(Channel channel) throws Exception {
                ChannelPipeline pipeline = channel.pipeline();
                Codec codec = Codec.EXTENSION_LOADER.getExtension(url.getProtocol());

                pipeline.addLast("frameDecoder",new NettyDecoder(codec));
                pipeline.addLast("frameEncoder", new NettyEncoder(codec));
                pipeline.addLast(new ReadTimeoutHandler(60));
                pipeline.addLast(new NettyServerHandler());
            }
        };
    }

    class NettyServerHandler extends ChannelDuplexHandler {

        private Connection connection;

        @Override
        public void channelActive(ChannelHandlerContext ctx) throws Exception {
            Channel channel = ctx.channel();
            NettyClientConnection connection = new NettyClientConnection(channel);
            this.connection = connection;
            connectionManager.addConnection(connection);
            if (logger.isDebugEnabled()) {
                logger.info("The connection of " + channel.remoteAddress() + " -> " + channel.localAddress()
                    + " is established.");
            }
        }


        @Override
        public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
            if (msg instanceof Message) {
                logger.debug("ChannelRead ---- {} ", msg);
                Message message = (Message) msg;
                if(message.isRequest()){
                    message.markResponse();

                    Object request = message.getData();

                    MessageHandler messageHandler = handlerRegistry.lookup(message);

                    Channel channel = ctx.channel();
                    CompletableFuture future = messageHandler.handle(connection,request);

                    future.whenComplete((r,t)-> {
                        message.setData(r);
                        //channel.write(message);
                        channel.writeAndFlush(message);
                        //ctx.writeAndFlush(message);
                    });
                } else {
                    CompletableFuture completableFuture = RpcCompletableFuture.removeFuture(message);
                    if(completableFuture != null){
                        completableFuture.complete(((Message) msg).getData());
                    }
                }
            }
        }

        @Override
        public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
            Channel channel = ctx.channel();
            logger.info("exceptionCaught", cause);
            if (logger.isDebugEnabled()) {

                logger.debug("The connection of " + channel.remoteAddress() + " ->  " + channel.localAddress()
                    + " occur exception.");
            }
            //clean TODO
            if (channel.isActive()) {
                channel.close();
            }
        }

        @Override
        public void userEventTriggered(ChannelHandlerContext ctx, Object evt) throws Exception {
            logger.info("userEventTriggered:evt {}", evt);
        }

        @Override
        public void channelInactive(ChannelHandlerContext ctx) throws Exception {
            Channel channel = ctx.channel();
            connectionManager.remove(connection);
            if (logger.isDebugEnabled()) {
                logger.debug(
                    "The connection of " + channel.remoteAddress() + " -> " + channel.localAddress() + " is broken.");
            }
        }

        @Override
        public void write(ChannelHandlerContext ctx, Object msg, ChannelPromise promise) throws Exception {
            logger.debug("Write msg {}", msg);
            super.write(ctx, msg, promise);
        }
    }
}
